#include "vl_thread_pool.h"
#include "vl_buffer.h"

#include <chrono>

using namespace VL;

/**
 * +======================================================================================================================+
 * @Author: CimZzz
 * @Date: 2022-08-21 15:26:19
 * 
 * 代码分割块，以下内容涉及: 
 * 
 * VLThreadRunnable
 * 
 */

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:11:24
 * 
 * 构造方法
 * 
 * @param runnable 可执行回调
 */
VLThreadRunnable::VLThreadRunnable(Runnable runnable): buffer(nullptr), futurePtr(nullptr), runnable(runnable), futureRunnable(nullptr) {

}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:11:46
 * 
 * 构造方法
 * 
 * @param runnable 携带 Future 的可执行回调
 */
VLThreadRunnable::VLThreadRunnable(FutureRunnable runnable): buffer(nullptr), futurePtr(nullptr), runnable(nullptr), futureRunnable(runnable),
	runnableWithBuffer(nullptr), futureRunnableWithBuffer(nullptr) {

}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:11:24
 * 
 * 构造方法
 * 
 * @param runnable 可执行回调, 使用 Buffer
 */
VLThreadRunnable::VLThreadRunnable(RunnableWithBuffer runnable):  buffer(nullptr), futurePtr(nullptr), runnable(nullptr), futureRunnable(nullptr),
	runnableWithBuffer(runnable), futureRunnableWithBuffer(nullptr) {

}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:11:46
 * 
 * 构造方法
 * 
 * @param runnable 携带 Future 的可执行回调, 使用 Buffer
 */
VLThreadRunnable::VLThreadRunnable(FutureRunnableWithBuffer runnable):  buffer(nullptr), futurePtr(nullptr), runnable(nullptr), futureRunnable(nullptr),
	runnableWithBuffer(nullptr), futureRunnableWithBuffer(runnable) {

}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:09:54
 * 
 * 线程执行方法
 */
void VLThreadRunnable::threadRun() {
	if(runnable) {
		runnable();
	} else if(futureRunnable) {
		futureRunnable(this->futurePtr);
	} else if(runnableWithBuffer) {
		runnableWithBuffer(this->buffer);
	} else if(futureRunnableWithBuffer) {
		futureRunnableWithBuffer(this->futurePtr, this->buffer);
	}
}


/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:41:45
 * 
 * 中断
 */
void VLThreadRunnable::interrupt() {
	if(this->futurePtr.get() != nullptr) {
		this->futurePtr->interrupt();
	}
}

/**
 * +======================================================================================================================+
 * @Author: CimZzz
 * @Date: 2022-08-21 16:29:04
 * 
 * 代码分割块，以下内容涉及: 
 * 
 * ThreadFutureManager
 * 
 */

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 16:25:55
 * 
 * 构造函数
 */
ThreadFutureManager::ThreadFutureManager() {

}


/**
 * @Author: CimZzz
 * @Date: 2022-08-21 16:27:44
 * 
 * 获取数据, 否则抛出异常
 */
void ThreadFutureManager::getOrThrow() {
	for(auto iter = this->futureList.begin() ; iter != this->futureList.end() ; iter ++) {
		(*iter)->getOrThrow();
	}
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 16:26:58
 * 
 * 添加 Future
 * 
 * @param futurePtr Future 指针
 */
void ThreadFutureManager::addFuture(const std::shared_ptr<VLFuture>& futurePtr) {
	this->futureList.push_back(futurePtr);
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 16:27:09
 * 
 * 获取 Future
 * 
 * @param index Future 下标
 * 
 * @return Future 指针
 */
std::shared_ptr<VLFuture> ThreadFutureManager::getFuture(unsigned int index) {
	if(index >= this->futureList.size()) {
		return std::shared_ptr<VLFuture>(nullptr);
	}

	return this->futureList.at(index);
}




/**
 * +======================================================================================================================+
 * @Author: CimZzz
 * @Date: 2022-08-21 15:28:31
 * 
 * 代码分割块，以下内容涉及: 
 * 
 * VLTempThread
 * VLThreadPool
 * 
 */


/**
 * @Author: CimZzz
 * @Date: 2023-02-28 11:02:58
 * 
 * 构造函数, 临时线程需要切换为类, 走析构函数
 */
VLTempThread::VLTempThread(): isRunning(true), threadPtr(nullptr) {

}

/**
 * @Author: CimZzz
 * @Date: 2023-02-28 11:02:52
 * 
 * 析构函数
 */
VLTempThread::~VLTempThread() {
	if(this->threadPtr.get() != nullptr) {
		if(this->threadPtr->joinable()) {
			this->threadPtr->join();
		}
	}
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 01:07:12
 * 
 * 构造函数
 * 
 * @param threadCount 常驻线程数
 * @param tempCount 临时线程数
 * @param tempKeepSeconds 临时线程持续时间, 单位秒
 * 
 * @note 当并发任务数超过常驻线程时, 线程池会开辟临时线程数来应对任务, 临时线程会持续一段时间后被回收
 */
VLThreadPool::VLThreadPool(unsigned int threadCount, unsigned int tempCount, unsigned int tempKeepSeconds): 
	threadCount(threadCount), 
	tempCount(tempCount),
	tempKeepSeconds(tempKeepSeconds), 
	curTempCount(0), 
	isInit(false), 
	isRunning(true), 
	taskCount(0), 
	unprocessedTaskCount(0), 
	isPending(false) {

}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 01:08:04
 * 
 * 析构函数
 */
VLThreadPool::~VLThreadPool() {
	this->mutex.lock();
	this->isRunning = false;
	this->mutex.unlock();
	this->clearTask();
	this->cond.notify_all();
	for(auto threadIter = this->threadList.begin() ; threadIter != this->threadList.end() ; threadIter ++) {
		auto threadPtr = (*threadIter);
		if(threadPtr->joinable()) {
			threadPtr->join();
		}
	}
	for(auto threadIter = this->tempThreadList.begin() ; threadIter != this->tempThreadList.end() ; threadIter ++) {
		auto threadPtr = (*threadIter);
		if(threadPtr->threadPtr->joinable()) {
			threadPtr->threadPtr->join();
		}
	}
}



/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:40:01
 * 
 * 执行线程回调
 * 
 * @param runnable 线程回调
 */
void VLThreadPool::exec(const std::shared_ptr<VLThreadRunnable>& runnable) {
	if(!this->isRunning) {
		runnable->interrupt();
		return;
	}

	if(this->isPending) {
		this->mutex.lock();
		this->pendingQueue.push(runnable);
		this->mutex.unlock();
	} else {
		this->mutex.lock();
		this->taskCount ++;
		this->unprocessedTaskCount ++;
		this->runnableQueue.push(runnable);
		this->checkNeedTempThread();
		this->mutex.unlock();
		this->cond.notify_all();
	}
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-23 21:03:14
 * 
 * 添加常驻线程
 */
void VLThreadPool::addThread() {
	this->threadList.push_back(std::shared_ptr<std::thread>(new std::thread([&](){
		// 线程循环
		std::shared_ptr<Buffer> buffer(new Buffer);
		while(this->isRunning) {
			std::shared_ptr<VLThreadRunnable> runnable(nullptr);
			{
				std::unique_lock<std::mutex> locker(this->mutex);
				if(!this->isRunning) {
					return;
				}

				if(this->runnableQueue.empty()) {
					// 队列中不存在回调, 等待
					this->cond.wait(locker);
				}

				if(!this->runnableQueue.empty()) {
					// 队列中不存在回调, 等待
					runnable = this->runnableQueue.front();
					this->runnableQueue.pop();
					this->unprocessedTaskCount --;
				}
			}

			if(runnable.get() != nullptr) {
				buffer->reset();
				runnable->buffer = buffer.get();
				runnable->threadRun();
				runnable.reset();
				std::lock_guard<std::mutex> locker(this->mutex);
				this->taskCount --;
				if(this->taskCount == 0) {
					// 全部任务完成
					this->taskDoneCond.notify_all();
				}
			}
		}
	})));
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-23 21:05:50
 * 
 * 添加临时线程
 */
void VLThreadPool::addTempThread() {
	this->curTempCount ++;
	std::shared_ptr<VLTempThread> tempThreadPtr(new VLTempThread);
	tempThreadPtr->threadPtr = std::shared_ptr<std::thread>(new std::thread([&, tempThreadPtr](){
		// 线程循环
		std::shared_ptr<Buffer> buffer(new Buffer);
		auto timePoint = std::chrono::system_clock::now();
		while(this->isRunning) {
			std::shared_ptr<VLThreadRunnable> runnable(nullptr);
			{
				std::unique_lock<std::mutex> locker(this->mutex);

				RUNNABLE_FETCH:

				if(!this->isRunning) {
					break;
				}

				if(this->runnableQueue.empty()) {
					// 队列中不存在回调, 等待
					this->cond.wait_for(locker, std::chrono::seconds(1));
				}

				auto currentTimePoint = std::chrono::system_clock::now();
				if(!this->runnableQueue.empty()) {
					// 队列中不存在回调, 等待
					runnable = this->runnableQueue.front();
					this->runnableQueue.pop();
					this->unprocessedTaskCount --;
					timePoint = currentTimePoint;
				} else {
					auto diffTime = currentTimePoint - timePoint;
					auto duration = std::chrono::duration_cast<std::chrono::seconds>(diffTime);
					if(duration.count() >= this->tempKeepSeconds) {
						// 空闲超时
						break;
					} else {
						goto RUNNABLE_FETCH;
					}
					
				}
			}

			if(runnable.get() != nullptr) {
				buffer->reset();
				runnable->buffer = buffer.get();
				runnable->threadRun();
				runnable.reset();
				std::lock_guard<std::mutex> locker(this->mutex);
				this->taskCount --;
				if(this->taskCount == 0) {
					// 全部任务完成
					this->taskDoneCond.notify_all();
				}
			}
		}
		
		this->mutex.lock();
		this->curTempCount --;
		tempThreadPtr->isRunning = false;
		this->mutex.unlock();
	}));
	this->tempThreadList.push_back(tempThreadPtr);
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-23 21:45:05
 * 
 * 检查是否需要添加新的临时线程
 */
void VLThreadPool::checkNeedTempThread() {
	// 首先对不激活的临时线程进行回收
	for(auto tempIter = this->tempThreadList.begin() ; tempIter != this->tempThreadList.end() ; ) {
		auto tempPtr = (*tempIter);
		if(tempPtr->isRunning) {
			tempIter ++;
		} else {
			tempIter = this->tempThreadList.erase(tempIter);
		}
	}
	// 根据当前任务数决定是否扩容临时线程
	if(this->unprocessedTaskCount > this->threadCount) {
		// 任务数超过常驻线程数
		auto diff = this->unprocessedTaskCount - this->threadCount;
		uint32_t expandCount = diff;
		if(diff + this->curTempCount > this->tempCount) {
			// 超过最大的临时线程数
			expandCount = this->tempCount - this->curTempCount;
		}
		while(expandCount > 0) {
			this->addTempThread();
			expandCount --;
		}
	}
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:10:39
 * 
 * 初始化线程池
 * 
 */
void VLThreadPool::initThreadPool() {
	if(this->isInit) {
		return;
	}
	this->isInit = true;
	// 构建线程池
	for(unsigned int i = 0 ; i < this->threadCount ; i ++) {
		this->addThread();
	}
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:12:41
 * 
 * 执行线程回调
 * 
 * @param runnable 线程回调
 */
void VLThreadPool::exec(VLThreadRunnable::Runnable runnable) {
	auto ptr = std::shared_ptr<VLThreadRunnable>(new VLThreadRunnable(runnable));
	this->exec(ptr);
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:12:41
 * 
 * 执行线程回调
 * 
 * @param runnable 线程回调
 */
void VLThreadPool::exec(VLThreadRunnable::RunnableWithBuffer runnable) {
	auto ptr = std::shared_ptr<VLThreadRunnable>(new VLThreadRunnable(runnable));
	this->exec(ptr);
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:58:00
 * 
 * 等待全部任务完成
 */
void VLThreadPool::waitForAllTaskDone() {
	if(!this->isRunning) {
		return;
	}
	std::unique_lock<std::mutex> locker(this->mutex);
	if(this->taskCount == 0) {
		return;
	}
	this->taskDoneCond.wait(locker);
}


/**
 * @Author: CimZzz
 * @Date: 2022-08-21 17:07:33
 * 
 * 将后续任务添加到等待队列中
 */
void VLThreadPool::pendingTask() {
	this->isPending = true;
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 17:07:51
 * 
 * 将等待队列中的任务放置到主消息队列中
 */
void VLThreadPool::continueTask() {
	if(this->isPending) {
		this->isPending = false;
		std::lock_guard<std::mutex> locker(this->mutex);
		while(!this->pendingQueue.empty()) {
			auto runnable = this->pendingQueue.front();
			this->pendingQueue.pop();
			this->runnableQueue.push(runnable);
			this->taskCount ++;
			this->unprocessedTaskCount ++;
		}
		this->checkNeedTempThread();
		this->cond.notify_all();
	}
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:14:57
 * 
 * 中断线程池中其他任务
 */
void VLThreadPool::clearTask() {
	std::lock_guard<std::mutex> locker(this->mutex);
	while(!this->runnableQueue.empty()) {
		auto runnable = this->runnableQueue.front();
		runnable->interrupt();
		this->runnableQueue.pop();
		this->taskCount --;
		this->unprocessedTaskCount --;
	}
	while(!this->pendingQueue.empty()) {
		auto runnable = this->pendingQueue.front();
		runnable->interrupt();
		this->pendingQueue.pop();
	}
	if(this->taskCount == 0) {
		this->taskDoneCond.notify_all();
	}
}